package com.stress.testing.plugin.apache.rocketmq.interceptor;

import com.stress.testing.common.StressTesting;
import com.stress.testing.common.TraceContextHolder;
import com.stress.testing.common.interceptor.AbstractInterceptor;
import org.apache.rocketmq.common.message.Message;

import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Iterator;

import static com.stress.testing.common.conf.PluginConfigKey.DEFAULT_PLUGIN_APACHE_ROCKETMQ_TOPIC_PREFIX;
import static com.stress.testing.common.conf.PluginConfigKey.PLUGIN_APACHE_ROCKETMQ_TOPIC_PREFIX_KEY;

public class DefaultMQProducerSendInterceptor extends AbstractInterceptor {

    @Override
    public void before(Method method, Object[] args, Object zuper, Object thiz) {
        if (!StressTesting.enable(getAgentConfig(), getPluginConfig(), TraceContextHolder.getTraceContext())) {
            return;
        }

        if (args == null || args.length == 0) {
            return;
        }

        Object arg = args[0];

        if (arg == null) {
            return;
        }

        if (arg instanceof Message) {
            Message msg = (Message) arg;
            changeTopic(msg);
            return;
        }

        if (arg instanceof Collection) {
            Collection colle = (Collection) arg;
            Iterator iterator = colle.iterator();
            while (iterator.hasNext()) {
                Object obj = iterator.next();
                if (obj instanceof Message) {
                    Message msg = (Message) obj;
                    changeTopic(msg);
                }
            }
        }
    }

    private void changeTopic(Message msg) {
        String topicOrigin = msg.getTopic();
        String topicPrefix = getPluginConfig().get(PLUGIN_APACHE_ROCKETMQ_TOPIC_PREFIX_KEY,
                DEFAULT_PLUGIN_APACHE_ROCKETMQ_TOPIC_PREFIX);

        if (topicOrigin == null || topicOrigin.startsWith(topicPrefix)) {
            return;
        }

        String newTopic = topicPrefix + topicOrigin;
        msg.setTopic(newTopic);
    }
}
